package com.opdar.athena.message.agent.client;

import com.opdar.athena.base.MessageType;
import com.opdar.athena.base.RedisSessionManager;
import com.opdar.athena.entities.TokenEntity;
import com.opdar.athena.message.agent.TransferServer;
import com.opdar.athena.messages.UserMessage;
import com.opdar.athena.tcp.TcpRecive;
import io.netty.channel.Channel;
import io.netty.channel.EventLoopGroup;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.UUID;

/**
 * Created by shiju on 2017/7/2.
 */
public class Client {
    private final InetSocketAddress address;
    private final EventLoopGroup group;
    private final Channel channel;
    private boolean isClose;
    private final Logger logger = LoggerFactory.getLogger(getClass());
    private long size;

    public Client(final InetSocketAddress address, final EventLoopGroup group, final Channel channel) {
        this.group = group;
        this.channel = channel;
        this.address = address;
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    logger.info("IP:{},端口:{}上线",address.getAddress().getHostAddress(),address.getPort());
                    channel.closeFuture().sync();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }finally {
                    isClose = true;
                    group.shutdownGracefully();
                    logger.info("IP:{},端口:{} 已断开连接.",address.getAddress().getHostAddress(),address.getPort());
                }
            }
        }).start();
    }

    public InetSocketAddress getAddress() {
        return address;
    }

    public Client messageAgentRegist() {
        UserMessage userMessage = new UserMessage();
        userMessage.setCommand("MessageAgentRegist");
        userMessage.setToken(UUID.randomUUID().toString());
        TcpRecive recive = new TcpRecive(MessageType.USER, userMessage);
        channel.writeAndFlush(recive);
        return this;
    }

    public void transfer(String messageId, TokenEntity entity) {
        ByteArrayOutputStream baos = null;
        DataOutputStream dataOutputStream = new DataOutputStream(baos = new ByteArrayOutputStream());
        try {
            byte[] messageData = messageId.getBytes();
            byte[] tokenData = entity.getToken().getBytes();
            dataOutputStream.writeInt(messageData.length);
            dataOutputStream.write(messageData);
            dataOutputStream.writeInt(tokenData.length);
            dataOutputStream.write(tokenData);
        } catch (IOException e) {
            e.printStackTrace();
        }
        TcpRecive recive = new TcpRecive(MessageType.TRANSFER, baos.toByteArray());
        try {
            channel.writeAndFlush(recive).sync().channel().read();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void heartbeat() {
        long time = System.currentTimeMillis();
        byte[] writeBuffer = new byte[8];
        writeBuffer[0] = (byte) (time >> 56);
        writeBuffer[1] = (byte) (time >> 48);
        writeBuffer[2] = (byte) (time >> 40);
        writeBuffer[3] = (byte) (time >> 32);
        writeBuffer[4] = (byte) (time >> 24);
        writeBuffer[5] = (byte) (time >> 16);
        writeBuffer[6] = (byte) (time >> 8);
        writeBuffer[7] = (byte) (time);
        TcpRecive recive = new TcpRecive(MessageType.HEARTBEAT, writeBuffer);
        channel.writeAndFlush(recive);
    }

    public boolean isOpen(){
        return !isClose;
    }

    public void migrate(String destination,String id, int type, long num, String token) {
        ByteArrayOutputStream baos = null;
        DataOutputStream dataOutputStream = new DataOutputStream(baos = new ByteArrayOutputStream());
        try {
            dataOutputStream.writeUTF(id);
            dataOutputStream.writeUTF(destination);
            dataOutputStream.writeInt(type);
            dataOutputStream.writeLong(num);
            dataOutputStream.writeUTF(token);
            TcpRecive recive = new TcpRecive(MessageType.MIGRATE, baos.toByteArray());
            channel.writeAndFlush(recive);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    public void setSize(long size) {
        this.size = size;
        RedisSessionManager redisSessionManager = TransferServer.getServer().getContext().getBean(RedisSessionManager.class);
        redisSessionManager.setClientCount(address.getAddress().getHostAddress()+":"+address.getPort(),size);
    }

    public long getSize() {
        return size;
    }
}
